import os
import sys

from kafka import KafkaProducer
import json


if __name__ == '__main__':
    kafka_broker, kafka_topic = sys.argv[1], sys.argv[2]
    files_dir = sys.argv[3]
    producer = KafkaProducer(bootstrap_servers=kafka_broker, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
    files = os.listdir(files_dir)
    for file in files:
        if file.endswith(".json"):
            with open(f'{files_dir}/{file}', 'r') as f:
                data = json.load(f)
            producer.send(kafka_topic, data)
            producer.flush()
    producer.close()
